package org.arl.modem.transportARQ;

import jade.core.AID;
import jade.core.behaviours.WakerBehaviour;

import java.util.Iterator;
import java.util.List;

import org.arl.modem.FAPIAgent;
import org.arl.modem.FAPIMessage;
import org.arl.modem.InputPDU;
import org.arl.modem.OutputPDU;

@SuppressWarnings("serial")
public class TransportARQ extends FAPIAgent {
	// declare all the constants used
	// Params - should be configurable
	private int PROTOCOL = 0; // used for link layer
	private int SCHEME = 0; // used for link layer
	private int RETRANSMISSION_TIME = 1000000; // for wake up behaviour
	private int PACKET_SIZE = 2;
	private int BLOCK_SIZE = 3; // number of packets in a block

	private int s = 3;
	private int sMax = 4;
	private int rxBlockSize;
	private int[] rxPacketIds;
	private int rxBlockStarted = 0;
	private int rxPacketCount = 0;
	// private int physicalBuffer = 3;
	// (timer)
	public final static String NOTIFICATIONS = "tsp-ntf";
	// FIXME :Rohit :08-Oct-2011 : add message id so that ACK can be checked for
	// the specific data packet

	// constants
	private final int MESSAGETYPE_DATA = 0;
	private final int MESSAGETYPE_ACK = 1;

	private AID lnk;
	private AID tsp;
	private AID phy;
	private AID notify;
	// private AID notify;
	// private int myAddress;
	private int totalPackets;
	// private byte[][] data;
	private int txStarted;
	private int sentPackets;
	private int ackReceived;
	// private int testAckcounter = 0;
	private int sentPacketCounter = 0;
	private WakerBehaviour timer;
	private WakerBehaviour rxTimer;
	private int JARQ_DELTA = 1200;

	private WakerBehaviour jarq;
	private PacketFactory factory;
	private List<Packet> packetsInTransmission;
	private Packet currentPacket;
	private Iterator<Packet> itrPackets;
	private boolean blockSend;
	private int blockPacketNumber;
	private FAPIMessage currentMessage;
	private FAPIMessage jarqmsg;

	public void setup() {
		super.setup();
		subscribeTopic("lnk-ntf");
		subscribeTopic("tsp-ntf");
		subscribeTopic("phy-ntf");
		// this.notify = getTopic("tsp-ntf");
		notify = getTopic(NOTIFICATIONS);
		this.lnk = new AID(getAgentProperty("lnk", "lnk"), AID.ISLOCALNAME);
		this.tsp = new AID(getAgentProperty("tsp", "tsp"), AID.ISLOCALNAME);
		this.phy = new AID(getAgentProperty("phy", "phy"), AID.ISLOCALNAME);
		this.txStarted = 0;
		// create and add behaviour as timer for given time

		info("$Id: TransportSW agent $");
	}

	/**
	 * main entry point for agent for any data or ACK messages
	 */
	protected void handleMessage(FAPIMessage msg) {

		/**
		 * check if the message is from Link layer
		 */
		// System.out.println("Sender is :" + msg.getSender());
		if (msg.getSender().equals(this.phy)) {
			switch (msg.getPerformative()) {
			case FAPIMessage.INFORM:
				if (msg.get("type").equals("TX.PKT.TIME")) {
					blockSend = false;
					// System.out.println("block pkt number:" +
					// blockPacketNumber
					// + "  s:" + s + "  block size:" + BLOCK_SIZE
					// + blockPacketNumber == BLOCK_SIZE + "");

					if (blockPacketNumber == BLOCK_SIZE && s > 0) {
						System.out.println("inside IF");
						// final FAPIMessage msgNext = createMessage(
						// FAPIMessage.REQUEST, tsp);
						// TransportPacket tp1 = new TransportPacket();
						// tp1.decode(currentMessage.getData());
						// // System.out.println("inside IF2");
						// msgNext.add("src", 1);
						// msgNext.add("dest", 2);
						// // FIXME :Rohit :08-Oct-2011 :remove hardcoded 2 from
						// // here
						// // msgNext.add("len", tp.messageLength / 2);
						// msgNext.add("tx", 1);

						jarq = new WakerBehaviour(this, JARQ_DELTA) {
							@Override
							protected void handleElapsedTimeout() {

								sendNextBlock(jarqmsg);
							}
						};
						// System.out.println("inside IF3");
						// System.out.println("adding jade behaviour with s=" +
						// s);
						addBehaviour(jarq);
						s--;
					} else if (blockPacketNumber < BLOCK_SIZE
							&& blockPacketNumber != 0) {
						sendNextPacket();
					}

				}
				// System.out.println("msg type:" + msg.get("type"));
				break;
			}
		}
		if (msg.getSender().equals(this.lnk)) {
			switch (msg.getPerformative()) {
			case FAPIMessage.INFORM:
				// System.out.println("incoming msg from link layer with source"
				// + msg.getAsInteger("source"));
				if (msg.getSender().equals(lnk)) {
					handleRxFAPI(msg);
					System.out.println("recvd pkt from link");

				}
				break;
			}
		}

		/**
		 * check if its sent from Transport layer
		 */
		if (msg.containsData()) {
			if (msg.getAsInteger("tx") != null && msg.getAsInteger("tx") == 1) {
				// System.out.println("outgoing msg from transport layer");
				handleTxFAPI(msg);
			}
		} else {
			handleRequest(msg);
		}

	}

	private void handleRequest(FAPIMessage msg) {
		String action = msg.getContent();
		if (action == null) {
			action = "";
		}
		FAPIMessage rsp = msg.createReply();
		rsp.setPerformative(FAPIMessage.NOT_UNDERSTOOD);
		if (action.equals(PARAM_SET)) { // param set request
			int ppp = msg.getAsInteger("parameter");
			int vvv = msg.getAsInteger("value");
			boolean err = false;
			switch (ppp) {
			case TransportParam.PACKET_SIZE:
				PACKET_SIZE = vvv;
				break;
			case TransportParam.PROTOCOL:
				PROTOCOL = vvv;
				break;
			case TransportParam.RETRANSMISSION_TIME:
				RETRANSMISSION_TIME = vvv;
				break;
			case TransportParam.SCHEME:
				SCHEME = vvv;
				break;
			case TransportParam.BLOCK_SIZE:
				BLOCK_SIZE = vvv;
				break;
			default:
				err = true;
			}
			if (err)
				rsp.setPerformative(FAPIMessage.FAILURE);
			else {
				rsp.setPerformative(FAPIMessage.INFORM);
				rsp.setContent(PARAM_SET);
				rsp.add("parameter", ppp);
				rsp.add("value", vvv);
			}
		} else if (action.equals(PARAM_GET)) { // param get request
			int ppp = msg.getAsInteger("parameter");
			long vvv = -1;
			switch (ppp) {
			case TransportParam.PACKET_SIZE:
				vvv = PACKET_SIZE;
				break;
			case TransportParam.PROTOCOL:
				vvv = PROTOCOL;
				break;
			case TransportParam.RETRANSMISSION_TIME:
				vvv = RETRANSMISSION_TIME;
				break;
			case TransportParam.SCHEME:
				vvv = SCHEME;
				break;
			case TransportParam.BLOCK_SIZE:
				vvv = BLOCK_SIZE;
				break;

			}
			if (vvv < 0) {
				rsp.setPerformative(FAPIMessage.FAILURE);
			}

			else {
				rsp.setPerformative(FAPIMessage.INFORM);
				rsp.setContent(PARAM_GET);
				rsp.add("parameter", ppp);
				rsp.add("value", vvv);
			}
		}
		send(rsp);
	}

	/**
	 * Handles the received FAPI message (incoming)
	 * 
	 * @param msg
	 */
	private void handleRxFAPI(FAPIMessage msg) {

		// call doRx() with messageType (Data(0) or ACK(1))
		doRx(msg);

	}

	/**
	 * Performs the actual Rx
	 * 
	 * @param tp
	 */
	private void doRx(FAPIMessage msg) {
		final TransportPacket tp = new TransportPacket();
		// decode the received packet
		tp.decode(msg.getData());
		FAPIMessage ntf = createMessage(FAPIMessage.INFORM, notify);
		ntf.add("src", tp.src);
		ntf.add("dest", tp.dest);
		ntf.add("RX", 1);

		ntf.setData(tp.messageData);

		switch (tp.messageType) {
		case 0: // Data Packet received
			/**
			 * if data packet received create and send the ACK
			 */
			// check if 1st packet in Blocked Rx
			ntf.add("TYPE", "RX.DATA");
			System.out.println("data Packet received with packetID: "
					+ tp.packetId);
			BLOCK_SIZE = tp.blockSize;
			if (rxBlockStarted == 0) {

				rxBlockSize = tp.blockSize;
				rxBlockStarted = 1;
				rxPacketCount = 0;
				rxPacketIds = new int[rxBlockSize];
				// FIXME :Rohit :Oct 23, 2011 :Start the timer based on block
				// size
				rxTimer = new WakerBehaviour(this, RETRANSMISSION_TIME
						* BLOCK_SIZE) {
					protected void handleElapsedTimeout() {
						sendAckMessage(tp.src, tp.dest);
						// System.out.println("Partial ACK sent");

					}
				};
				addBehaviour(rxTimer);
				sentPacketCounter = 0;
			}
			rxPacketIds[rxPacketCount] = tp.packetId;
			rxPacketCount++;
			// System.out.println("Rx packet id:" + tp.packetId
			// + ":Rx packet count:" + rxPacketCount
			// + ":and block size is:" + rxBlockSize);
			// OutputPDU pdu = new OutputPDU(1);
			// pdu.write8(tp.packetId);
			// rxPacketIds.add(new Byte(pdu.toByteArray()[0]));

			// testAckcounter++;
			// String sdat = new String(tp.messageData);
			// System.out.println("Data received with packet ID:" + tp.packetId
			// + "is :" + sdat);

			if (rxPacketCount == rxBlockSize) {
				removeBehaviour(rxTimer);
				sendAckMessage(tp.src, tp.dest);
			}
			break;
		case 1: // ACK packet received
			/**
			 * if ACK received check if any more packets are available if yes
			 * then check if it is the last packet if yes send the next packet
			 * and clear counter else just send the packet and increase counters
			 */
			ntf.add("TYPE", "RX.ACK");
			// stop the timer
			timer.stop();
			removeBehaviour(timer);

			InputPDU inpdu = new InputPDU(tp.messageData);
			// System.out.println("ACK's tp.msgfata length:"
			// + tp.messageData.length + ":in pdu's length:"
			// + inpdu.length());
			;
			for (int k = 0; k < inpdu.length(); k++) {
				int ackPacketId = inpdu.read8u();
				// System.out.println("Packet number ACK received" +
				// ackPacketId);
				factory.ackPacket(ackPacketId);
			}

			// increase the ACK counter
			ackReceived++;
			// System.out.println("ACK received:" + ackReceived +
			// "Total packets"
			// + totalPackets);

			// all packets are not sent yet
			if (ackReceived < totalPackets / BLOCK_SIZE) {
				// create FAPIMESSAGE to send DATA via Transport layer
				FAPIMessage msgNext = createMessage(FAPIMessage.REQUEST, tsp);
				msgNext.add("src", tp.dest);
				msgNext.add("dest", tp.src);
				// FIXME :Rohit :08-Oct-2011 :remove hardcoded 2 from here
				// msgNext.add("len", tp.messageLength / 2);
				msgNext.add("tx", 1);
				// sendNext(msgNext);
				sendNextBlock(msgNext);
			} else {
				// last ACK received
				// FIXME :Rohit :08-Oct-2011 : change the ACK behaviour
				txStarted = 0;
			}
			break;
		default:
			// System.out.println("msg type before error:" + tp.messageType);
			// System.out.println("error in RX");
			break;
		}
		send(ntf);

	}

	private void sendAckMessage(int src, int dest) {
		// TODO Auto-generated method stub
		rxBlockStarted = 0;
		rxPacketCount = 0;
		// create a FAPIMESSAGE to send the ACK via LINK layer
		FAPIMessage ackMessage = createMessage(FAPIMessage.REQUEST, this.lnk);
		ackMessage.add("destination", src);
		ackMessage.add("protocol", PROTOCOL);
		ackMessage.add("scheme", SCHEME);
		// Create Transport layer ACK packet

		TransportPacket ackPacket = new TransportPacket();
		ackPacket.messageType = MESSAGETYPE_ACK;
		// FIXME :Rohit :Oct 20, 2011 : mesage length may not be proper
		ackPacket.messageLength = rxPacketIds.length;
		ackPacket.blockSize = BLOCK_SIZE;
		ackPacket.src = dest;
		ackPacket.dest = src;

		// System.out.println("rx pktids length:" + rxPacketIds.length);
		OutputPDU pdu = new OutputPDU(rxPacketIds.length);
		for (int k : rxPacketIds) {
			System.out.println("writing ack for pkt id:" + k);
			pdu.write8(k);
		}

		ackPacket.messageData = pdu.toByteArray();
		// System.out.println("Data Packet received with source:" +
		// tp.src
		// + "and destination:" + tp.dest);
		// encode the Transport packet
		byte[] buf = ackPacket.encode();

		ackMessage.setData(buf);

		// send the ACK
		// if (testAckcounter % 2 == 0)
		send(ackMessage);
	}

	private void handleTxFAPI(FAPIMessage msg) {
		/**
		 * if Transmission is not yet started Break the packet to smaller chunks
		 * and send the 1st packet
		 */
		if (this.txStarted == 0) {
			// start transmission session and reset the counters
			this.txStarted = 1;
			sentPackets = 0;
			ackReceived = 0;
			// data = null;
			totalPackets = 0;
			// s = 3;
			// convert the data into number of packets based on packet size
			byte[] buf = new byte[msg.getData().length];
			buf = msg.getData();
			// data = new byte[msg.getData().length / PACKET_SIZE][PACKET_SIZE];
			// number of packets to be transmitted in current session

			// System.out.println("PAcket size(in bytes):" + PACKET_SIZE);
			// System.out.println("Block size(in number of pkts):" +
			// BLOCK_SIZE);

			factory = new PacketFactory(PACKET_SIZE, BLOCK_SIZE);
			totalPackets = factory.createPackets(buf);
			// System.out.println("total packets are:" + totalPackets);
			// for (int i = 0; i < totalPackets; i++) {
			// for (int k = 0; k < PACKET_SIZE; k++) {
			// System.out.println("data part:" + buf[i * PACKET_SIZE + k]);
			// data[i][k] = buf[i * PACKET_SIZE + k];
			// }
			//
			// }

			msg.setData(null);
			jarqmsg = msg;
			// delegate the transmission to sendNext
			sendNextBlock(msg);
			// try {
			// Thread.sleep(1);
			// } catch (InterruptedException e) {
			// // TODO Auto-generated catch block
			// e.printStackTrace();
			// }
			s--;
		} else {
			// FIXME :Rohit :08-Oct-2011 : error transmission is in progress
		}

	}

	/**
	 * Send Next Block of packets
	 */
	private void sendNextBlock(FAPIMessage msg) {
		// FIXME :Rohit :Oct 20, 2011 : check for end of stream and adjust the
		// block size
		System.out.println("Sending the next block");
		if (packetsInTransmission != null) {
			packetsInTransmission.clear();
		}

		// if (s == 0) {
		// packetsInTransmission = factory.getNextBlock(1);
		// } else {
		if (s == 0) {
			packetsInTransmission = factory.getNextBlock(sMax - 1);

		} else {
			packetsInTransmission = factory.getNextBlock(sMax - s);

		}
		// }

		blockPacketNumber = 0;
		itrPackets = packetsInTransmission.iterator();

		// send the 1st packet and block the block send
		currentMessage = msg;
		sendNextPacket();

	}

	private void sendNextPacket() {
		sentPacketCounter++;
		blockPacketNumber++;
		sendNext(currentMessage);
		blockSend = true;

	}

	/**
	 * to send the next available packet/message , called to transmit the next
	 * packet
	 * 
	 * @param msg
	 */
	private void sendNext(FAPIMessage msg) {

		if (sentPackets / BLOCK_SIZE == ackReceived) {// there is no error

			if (sentPackets < totalPackets) { // still packets are available

				currentPacket = itrPackets.next();
				// System.out.println("Sending the packet ID:"
				// + currentPacket.getPacketId());
				msg.setData(currentPacket.getData());
				handleTx(msg);
			} else {
				// FIXME :Rohit :08-Oct-2011 :indicate the completion of
				// transmission session
				System.out.println("completed");
			}

		} else { // in case of error retransmission of older packet
			// msg.setData(data[sentPackets - 1]);
			currentPacket = itrPackets.next();
			msg.setData(currentPacket.getData());
			handleTx(msg);
		}

	}

	/**
	 * Handles the Tx operation
	 * 
	 * @param msg
	 */
	private void handleTx(FAPIMessage msg) {
		// create a reply for command handler
		FAPIMessage rsp = msg.createReply();
		rsp.setPerformative(FAPIMessage.FAILURE);
		if (doTx(msg)) { // if the transmission is successful
			rsp.setPerformative(FAPIMessage.AGREE);
		}
		send(rsp);
	}

	/**
	 * perform actual transmission
	 * 
	 * @param msg
	 * @return
	 */
	private boolean doTx(final FAPIMessage msg) {
		// create the Transport packet

		TransportPacket tp = new TransportPacket();
		tp.messageType = MESSAGETYPE_DATA;
		tp.messageLength = msg.getData().length; // msg.getAsInteger("len");
		tp.blockSize = BLOCK_SIZE;
		tp.blockId = 1;
		tp.messageData = msg.getData();
		tp.src = msg.getAsInteger("src");
		tp.dest = msg.getAsInteger("dest");
		// System.out.println("in doTx() packetID:" +
		// currentPacket.getPacketId()
		// + "with data:" + new String(currentPacket.getData())
		// + "messgae.getdata:" + msg.getData());
		tp.packetId = currentPacket.getPacketId();
		// encode the transport packet
		byte[] buf = tp.encode();
		// TransportPacket tpdecode = new TransportPacket();
		// tpdecode.decode(buf);
		System.out.println("Sending packet ID:" + tp.packetId);
		// + "with data:" + new String(tpdecode.messageData));
		// Create a FAPIMESSAGE to send the data via LINK layer (with data as
		// encoded Transport packet)
		FAPIMessage req = createMessage(FAPIMessage.REQUEST, this.lnk, buf);
		// System.out.println("encoded packet from transport layer:"
		// + new String(buf));
		req.add("destination", tp.dest);
		req.add("protocol", PROTOCOL);
		req.add("scheme", SCHEME);
		send(req);
		sentPackets++;

		// FIXME :Rohit :Oct 19, 2011 :add the timer only when block ack is
		// expected
		if (sentPacketCounter == BLOCK_SIZE) {
			timer = new WakerBehaviour(this, RETRANSMISSION_TIME) {
				protected void handleElapsedTimeout() {
					// perform operation X

					// int re = sentPackets - BLOCK_SIZE;
					// System.out.println("Retransmission of whole block");
					msg.setData(null);
					// sendNextBlock(msg);
				}
			};
			addBehaviour(timer);
			sentPacketCounter = 0;
		}

		return true;
	}

}
